package dataClean;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * @author legolas
 * @date 2020/3/14 下午4:27
 * 1.从原始数据(JSON格式)中过滤出需要的字段
 * 2.针对核心字段进行异常值判断 （自己定义规则，例如：如果为负数或者缺失，则认为时异常数据，直接丢弃）
 */
public class DataCleanJob {

    public static void main(String[] args) {
        try {
            if (args.length != 2) {
                //如果参数不够，程序直接退出
                System.out.print("请指定输入路径和输出路径！");
                System.exit(100);
            }
            //创建Job需要的配置参数
            Configuration conf = new Configuration();
            //创建一个job
            Job job = Job.getInstance(conf);

            //注意：这一行必须设置，否则在集群中执行找不到WordCountJob这个类
            job.setJarByClass(DataCleanJob.class);

            //指定输入路径
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            //指定map相关代码
            job.setMapperClass(DataCleanMap.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(LongWritable.class);

            //reduce禁用
            job.setNumReduceTasks(0);

            //提交job
            job.waitForCompletion(true);

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

